package org.apache.ocean;

//import org.apache.lucene.index.*;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.logging.Logger;

import org.apache.lucene.search.Query;
import org.apache.lucene.search.Similarity;
import org.apache.lucene.search.Weight;

/**
 *
 * @author jasonr
 */
// TODO: implement unit tests of MultiShardSearch
public class MultiShardSearch {
	public static Logger log = Logger.getLogger(MultiShardSearch.class.getName());
  Query rewrittenQuery;
  long offset;
  long length;
  List<Query> filterQueries;
  List<SearchIndexShard> searchIndexShards;
  List<ShardSearch> shardSearches;
  Weight weight;
  Similarity similarity;
  int maxDoc;
  TimeoutState timeoutState;
  CLong timeout;
  ThreadPoolExecutor threadPool;
  TermFrequencies termFrequencies;
  FacetResponse facetResponse;
  IndexSchema indexSchema;
  
  public MultiShardSearch(Query rewrittenQuery, long offset, long length, List<Query> filterQueries, FacetResponse facetResponse, List<SearchIndexShard> searchIndexShards, Similarity similarity, TimeoutState timeoutState, TermFrequencies termFrequencies, ThreadPoolExecutor threadPool, IndexSchema indexSchema) throws Exception {
    this.rewrittenQuery = rewrittenQuery;
    this.offset = offset;
    this.length = length;
    this.filterQueries = filterQueries;
    this.searchIndexShards = searchIndexShards;
    this.similarity = similarity;
    this.timeout = timeout;
    if (termFrequencies == null) throw new IllegalArgumentException("termFrequencies null");
    this.termFrequencies = termFrequencies;
    this.facetResponse = facetResponse;
    this.threadPool = threadPool;
    this.timeoutState = timeoutState;
    this.indexSchema = indexSchema;
    shardSearches = new ArrayList<ShardSearch>(searchIndexShards.size());
    maxDoc = 0;
    for (SearchIndexShard searchIndexShard : searchIndexShards) {
      ShardSearch shardSearch = new ShardSearch(0, offset+length, filterQueries, facetResponse, searchIndexShard, timeoutState, indexSchema);
      maxDoc += shardSearch.maxDoc();
      shardSearches.add(shardSearch);
    }
    weight = createWeight(rewrittenQuery);
  }
  
  public SearchResult search(Distance distance) throws Exception, GeneralTimeoutException {
    CHitQueue hitQueue = new CHitQueue((int)length);
    List<SearchResult> searchResults = executeTasks(null, distance);
    CLong numFound = new CLong(0);
    for (SearchResult searchResult : searchResults) {
      numFound.value += searchResult.numFound.value;
      for (CScoreDoc scoreDoc : searchResult.docs) {
        if (!hitQueue.insert(scoreDoc)) {
        	break;
        }
      }
    }
    return new SearchResult(ClusterUtil.toScoreDocs((int)offset, (int)length, hitQueue), numFound, indexSchema);
  }
  
  private List<SearchResult> executeTasks(CSortFields sortFields, Distance distance) throws Exception, GeneralTimeoutException {
  	List<Callable<Object>> searchTasks = new ArrayList<Callable<Object>>(shardSearches.size());
    for (ShardSearch shardSearch : shardSearches) {
      searchTasks.add(new SearchTask(sortFields, distance, shardSearch));
    }
    return ClusterUtil.executeTasks(searchTasks, threadPool, timeoutState);
  }
  
  public SearchResult search(CSortFields sortFields, Distance distance) throws Exception, GeneralTimeoutException {
    FieldReaderSortedQueue queue = new FieldReaderSortedQueue(sortFields, (int)length+(int)offset);
    List<SearchResult> searchResults = executeTasks(sortFields, distance);
    CLong numFound = new CLong(0);
    for (SearchResult searchResult : searchResults) {
      for (CScoreDoc fieldDoc : searchResult.docs) {
      	if (fieldDoc != null) {
          if (!queue.insert(fieldDoc)) break;
      	}
      }
      numFound.value += searchResult.numFound.value;
    }
    return new SearchResult(ClusterUtil.toScoreDocs((int)offset, (int)length, queue), numFound, indexSchema);
  }
  
  public class SearchTask implements Callable {
    CSortFields sortFields;
    ShardSearch shardSearch;
    Distance distance;
    
    public SearchTask(CSortFields sortFields, Distance distance, ShardSearch shardSearch) {
      this.sortFields = sortFields;
      this.distance = distance;
      this.shardSearch = shardSearch;
    }
    
    public Object call() throws Exception {
      try {
        if (sortFields == null) {
          return shardSearch.search(weight, distance);
        } else {
          return shardSearch.search(weight, sortFields, distance);
        }
      } catch (Throwable throwable) {
        if (throwable instanceof Exception) throw (Exception)throwable;
        else throw new Exception("", throwable);
      }
    }
  }
  
  public int maxDoc() {
    return maxDoc;
  }
  
  public Weight createWeight(Query rewrittenQuery) throws Exception {
    return rewrittenQuery.weight(new CachedDocumentFrequencySearcher(termFrequencies, maxDoc, similarity));
  }
}
